package com.kpmg.datalake.common.service.impl;

import com.alibaba.fastjson.JSON;
import com.kpmg.datalake.common.constants.Constants;
import com.kpmg.datalake.common.enums.ResultCodeEnum;
import com.kpmg.datalake.common.enums.SystemParamsEnum;
import com.kpmg.datalake.common.enums.TopicEnum;
import com.kpmg.datalake.common.exception.BusinessException;
import com.kpmg.datalake.common.service.IComSystemParamsService;
import com.kpmg.datalake.common.service.IDbPrmsnService;
import com.kpmg.datalake.common.service.IIdyTabMpngService;
import com.kpmg.datalake.common.service.IProjectService;
import com.kpmg.datalake.common.service.SqlLoaderService;
import com.kpmg.datalake.common.utils.AssertUtil;
import com.kpmg.datalake.common.utils.EncrypUtil;
import com.kpmg.datalake.common.vo.KeyAndValueDTO;
import com.kpmg.datalake.db.model.DbPrmsn;
import com.kpmg.datalake.db.model.Project;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import javax.annotation.Resource;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

/**
 * @author Alex.zhao [alex.zhao@kpmg.com]
 * @since Created on azhao6 on 2018-08-21 23:06:38
 */
@Service
public class SqlLoaderServiceImpl implements SqlLoaderService {

  /**
   * 表数据分隔符
   */
  private static final String TABLE_DATA_SEPARATOR = "DATA_";
  /**
   * 表头分割服
   */
  private static final String TABLE_HEAD_SEPARATOR = "TABLE_";
  /**
   * 错误输出
   */
  public static final String ERROR_OUTPUT = "errorPut";
  /**
   * bad 输出
   */
  public static final String BAD_OUTPUT = "badPut";
  @Resource
  private Executor executor;

  private static Logger logger = LoggerFactory.getLogger(SqlLoaderServiceImpl.class);

  @Resource
  private IComSystemParamsService comSystemParamsService;

  @Resource
  private IIdyTabMpngService idyTabMpngService;
  @Resource
  private IProjectService projectService;
  @Resource
  private IDbPrmsnService dbPrmsnService;
  @Resource
  private CommonExecuteServiceImpl commonExecuteService;

  @Override
  public Map<String, Map<String, String>> loaderData(String projectId, List<String> fileList)
      throws IOException, InterruptedException, ExecutionException {
    Project project = projectService.selectById(projectId);
    //数据导之前的common库执行脚本
    dbPrmsnService.initIndustryInfo(project, true, TopicEnum.INIT_DATA);
    HashMap<String, KeyAndValueDTO<String, String>> dtoHashMap = this.fileListToTableList(fileList);
    //表头校验 看看是否同时存在一个数据文件和表头文件
    this.validateTableList(dtoHashMap);
    Map<String, Map<String, String>> resultMap = new HashMap<>();
    for (Entry<String, KeyAndValueDTO<String, String>> entry : dtoHashMap.entrySet()) {
      //加载数据
      Map<String, String> loadResult = this.loaderDataFromFile
          (projectId, entry.getKey(), entry.getValue().getValue(), entry.getValue().getKey());
      resultMap.put(entry.getKey(), loadResult);
    }
    //删除本地存储文件
    try {
      for (String fileName : fileList) {
        FileUtils.forceDelete(new File(fileName));
      }
    } catch (Exception e) {
      logger.error(e.getMessage(), e);
    }
    //索引重建
    rebuildIndex(projectId);
    //项目导入完成后 后续脚本
    dbPrmsnService.initIndustryInfo(project, false, null);
    //数据导完以后的common库执行脚本
    dbPrmsnService.initIndustryInfo(project, false, TopicEnum.INIT_DATA);
    return resultMap;
  }

  /**
   * 重建用户失效索引
   *
   * @param projectId projectId
   */
  private void rebuildIndex(String projectId) {
    try {
      List<Map<String, Object>> mapList = commonExecuteService.queryListByString(projectId,
          "select 'alter  index '||index_name ||' rebuild ' TEXT  from user_indexes where  status <>'VALID'");
      if (!CollectionUtils.isEmpty(mapList)) {
        for (Map<String, Object> sql : mapList) {
          Object text = sql.get("TEXT");
          if (text != null) {
            commonExecuteService.updateByString(projectId, String.valueOf(text));
          }
        }

      }
    } catch (Exception e) {
      logger.error("重建索引失败{}", e);
    }

  }

  /**
   * 从文件中加载数据到数据库
   *
   * @param projectId 项目ID 对于多数据源databaseId
   * @param tableName 表名
   * @param headFile 表头文件
   * @param dataFile 数据文件
   */
  @Override
  public Map<String, String> loaderDataFromFile(String projectId, String tableName, String headFile,
      String dataFile) throws IOException, InterruptedException, ExecutionException {
    //查询项目信息
    Project project = projectService.selectProjectAndEntAndDbById(projectId);
    //来源表
    String sourceTable = idyTabMpngService
        .getTargetTabBySourceTabAndIdyId(project.getEntityBasic().getIdyId(), tableName);

    return this.execSqlldr(projectId, sourceTable, headFile, dataFile);
  }


  /**
   * 生成sql control 文件
   *
   * @param projectId 项目名称
   * @param tableName 表名
   * @param headFileName 表头文件
   * @param dataFileName 数据文件
   * @return map 处理结果
   */
  private Map<String, String> execSqlldr(String projectId, String tableName, String headFileName,
      String dataFileName) throws IOException, InterruptedException, ExecutionException {
    HashMap<String, String> hashMap = new HashMap<>();
    //sqlldr 执行脚本
    String dataLakeSqlldrCmd = comSystemParamsService
        .getSystemParamsByCode(SystemParamsEnum.DATA_LAKE_SQL_LDR_CMD.getCode());
    AssertUtil.assertNotEmpty(dataLakeSqlldrCmd, "sqlload执行文件不能为空");
    //ctl 文件生成规则
    String dataLakeSqlldrCtl = comSystemParamsService
        .getSystemParamsByCode(SystemParamsEnum.DATA_LAKE_SQL_LDR_CTL.getCode());
    AssertUtil.assertNotEmpty(dataLakeSqlldrCtl, "控制文件生成规则不能为空");
    //file格式化样式
    String dataLakeSqlldrFieldFormat = comSystemParamsService
        .getSystemParamsByCode(SystemParamsEnum.DATA_LAKE_SQL_LDR_FIELD_FORMAT.getCode(), "%s");

    //file时间格式化格式化样式
    String dataLakeSqlldrFieldDateFormat = comSystemParamsService
        .getSystemParamsByCode(SystemParamsEnum.DATA_LAKE_SQL_LDR_FIELD_DATE_FORMAT.getCode(),
            "%s \"to_date(:%s ,'yyyy-mm-dd hh24:mi:ss')\" ");
    //表头分隔符
    String dataLakeSqlldrHeadSeparatorFlag = comSystemParamsService
        .getSystemParamsByCode(SystemParamsEnum.DATA_LAKE_SQL_LDR_HEAD_SEPARATOR_FLAG.getCode(),
            "\\|");

    Project project = projectService.selectProjectAndEntAndDbById(projectId);
    DbPrmsn database = project.getDatabase();

    File headFile = new File(headFileName);
    String fileParentPath = headFile.getParent();
    String headString = FileUtils.readFileToString(headFile, Charset.defaultCharset());
    StringBuilder builder = new StringBuilder();
    Set<String> dateColumn = dbPrmsnService.getDateColumn(projectId, tableName);

    for (String head : headString.split(dataLakeSqlldrHeadSeparatorFlag)) {
      builder.append(",");
      //去掉  回车 换行   tab 制表符 空格
      head = head.replaceAll("\\s|\t|\r|\n", "").toUpperCase();
      //判断表头
      if (dateColumn.contains(head)) {
        builder.append(String.format(dataLakeSqlldrFieldDateFormat, head, head, head));
      } else {
        builder.append(String.format(dataLakeSqlldrFieldFormat, head, head, head));
      }
    }
    dataLakeSqlldrCtl = String
        .format(dataLakeSqlldrCtl, dataFileName, tableName, builder.substring(1));
    String ctlFileName = fileParentPath + File.separator + tableName + ".ctl";
    String logFileName = fileParentPath + File.separator + tableName + ".log";
    String badDataName = fileParentPath + File.separator + tableName + ".bad";
    FileUtils.write(new File(ctlFileName), dataLakeSqlldrCtl, Charset.defaultCharset());
    //生成sql 文件

    String sqlldrCmd = String
        .format(dataLakeSqlldrCmd, this.getDbConnect(database), ctlFileName, logFileName,
            badDataName,
            dataFileName);

    KeyAndValueDTO<String, String> result = executiveSqlldr(sqlldrCmd);
    //记录错误日志
    try {
      File badFile = new File(badDataName);
      if (badFile.exists()) {
        hashMap.put(BAD_OUTPUT,
            FileUtils.readFileToString(new File(logFileName), Charset.defaultCharset()));
      }
    } catch (IOException e) {
      //do nothing ,for me this is success
    }
    hashMap.put(ERROR_OUTPUT, result.getValue());
    return hashMap;
  }

  /**
   * 根据操作系统获取中英文。
   *
   * @param database 数据库
   * @return 数据库连接信息
   */
  private String getDbConnect(DbPrmsn database) {
    //生成sql 文件
    StringBuilder builder = new StringBuilder();
    if (this.isWindows()) {
      builder.append(EncrypUtil.decryptAes(database.getDbAcc(), Constants.ENCRY_KEY));
      builder.append("/\"\"\"");
      builder.append(EncrypUtil.decryptAes(database.getDbPwd(), Constants.ENCRY_KEY));
      builder.append("\"\"\"");
    } else {
      builder.append("'");
      builder.append(EncrypUtil.decryptAes(database.getDbAcc(), Constants.ENCRY_KEY));
      builder.append("/'");
      builder.append(EncrypUtil.decryptAes(database.getDbPwd(), Constants.ENCRY_KEY));
      builder.append("'\"");
    }
    builder.append("@");
    builder.append(database.getIpAdr());
    builder.append(":");
    builder.append(database.getPort());
    builder.append("/");
    builder.append(database.getDbInstnnm());
    return builder.toString();
  }

  /**
   * 判断是否windows 操作系统
   *
   * @return 是否
   */
  private boolean isWindows() {
    return System.getProperty("os.name").toLowerCase().startsWith("win");
  }


  /**
   * final Process process=Runtime.getRuntime().exec(cmd,new String[]{"ORACLE_HOME=/app/oracle/product/12.1.0/db","LD_LIBRARY_PATH=/usr/local/lib:/app/oracle/product/12.1.0/db/lib:$LD_LIBRARY_PATH"});
   *
   * @param sqlldrCmd 需要执行的命令
   * @throws IOException IOException
   * @throws InterruptedException * @throws InterruptedException
   */
  public KeyAndValueDTO<String, String> executiveSqlldr(String sqlldrCmd)
      throws IOException, InterruptedException, ExecutionException {
    KeyAndValueDTO<String, String> keyAndValueDTO = new KeyAndValueDTO<>();
    String[] cmd;

    if (this.isWindows()) {
      cmd = new String[]{"cmd.exe", "/C", sqlldrCmd};
    } else {
      cmd = new String[]{"/bin/bash", "-c",
          "echo $ORACLE_HOME;echo $LD_LIBRARY_PATH;$ORACLE_HOME/bin/" + sqlldrCmd};
    }
    final Process process = Runtime.getRuntime().exec(cmd);
    // 处理InputStream的线程
    ((ThreadPoolTaskExecutor) executor).submit(() -> {
      BufferedReader in = new BufferedReader(new InputStreamReader(process.getInputStream()));
      StringBuilder builder = new StringBuilder();
      String line;
      try {
        while ((line = in.readLine()) != null) {
          builder.append(new String(line.getBytes(), Charset.defaultCharset()));
          logger.warn("output: {} ", new String(line.getBytes(), Charset.defaultCharset()));
        }
      } catch (IOException e) {
        logger.error(e.getMessage(), e);
      } finally {
        try {
          in.close();
        } catch (IOException e) {
          logger.error(e.getMessage(), e);

        }
      }
      return builder.toString();
    });
    // 处理ErrorStream的线程
    Future<String> error = ((ThreadPoolTaskExecutor) executor).submit(() -> {
      BufferedReader in = new BufferedReader(new InputStreamReader(process.getErrorStream()));
      String line;
      StringBuilder builder = new StringBuilder();
      try {
        while ((line = in.readLine()) != null) {
          builder.append(new String(line.getBytes(), Charset.defaultCharset()));
          logger.error("error:{} ", new String(line.getBytes(), Charset.defaultCharset()));
        }
      } catch (IOException e) {
        logger.error(e.getMessage(), e);
      } finally {
        try {
          in.close();
        } catch (IOException e) {
          logger.error(e.getMessage(), e);
        }
      }
      return builder.toString();
    });
    int exitValue = process.waitFor();
    String result = exitValue == 0 ? "成功" : "失败";
    logger.warn("返回值：{}\n数据导入{}", exitValue, result);
    process.getOutputStream().close(); // 关闭
    keyAndValueDTO.setValue(error.get());
    return keyAndValueDTO;
  }

  /**
   * 处理后的表数据内容校验
   *
   * @param dtoHashMap 表头信息
   */
  private void validateTableList(HashMap<String, KeyAndValueDTO<String, String>> dtoHashMap) {
    List<KeyAndValueDTO<String, String>> errList = new ArrayList<>();
    for (KeyAndValueDTO<String, String> kav : dtoHashMap.values()) {
      if (StringUtils.isBlank(kav.getKey()) || StringUtils.isBlank(kav.getValue())) {
        errList.add(kav);
      }
    }
    if (!CollectionUtils.isEmpty(errList)) {
      throw new BusinessException(ResultCodeEnum.CODE4000,
          String.format("存在缺少表头或表文件的表 %s", JSON.toJSONString(errList)));
    }

  }

  /**
   * 文件列表转成 表头 列表
   *
   * @param fileList 文件信息 key 数据文件  value 是表头问题
   * @return 表头列表
   */
  private HashMap<String, KeyAndValueDTO<String, String>> fileListToTableList(
      List<String> fileList) {
    HashMap<String, KeyAndValueDTO<String, String>> dtoHashMap = new HashMap<>();
    for (String filePath : fileList) {
      //文件名称
      String fileName = filePath
          .substring(filePath.lastIndexOf(File.separator) + File.separator.length());
      if (fileName.lastIndexOf('.') == -1) {
        continue;
      }
      fileName = fileName.substring(0, fileName.lastIndexOf('.'));
      KeyAndValueDTO<String, String> tabFileList;
      //数据文件
      if (fileName.startsWith(TABLE_DATA_SEPARATOR)) {
        fileName = fileName.substring(TABLE_DATA_SEPARATOR.length());
        tabFileList = this.getTabFileName(dtoHashMap, fileName);
        tabFileList.setKey(filePath);
        dtoHashMap.put(fileName, tabFileList);
        //表头文件
      } else if (fileName.startsWith(TABLE_HEAD_SEPARATOR)) {
        fileName = fileName.substring(TABLE_HEAD_SEPARATOR.length());
        tabFileList = this.getTabFileName(dtoHashMap, fileName);
        tabFileList.setValue(filePath);
        dtoHashMap.put(fileName, tabFileList);
      } else {
        logger.error("unknown filePath has data or head :{}", filePath);
      }
    }
    return dtoHashMap;
  }

  /**
   * 根据 文件表名获取文件
   *
   * @param dtoHashMap 原有数据
   * @param fileName 文件表名
   * @return 对象信息
   */
  private KeyAndValueDTO<String, String> getTabFileName(
      Map<String, KeyAndValueDTO<String, String>> dtoHashMap, String fileName) {
    KeyAndValueDTO<String, String> result = new KeyAndValueDTO<>();
    if (dtoHashMap == null) {
      dtoHashMap = new HashMap<>(2 >> 4);
    }
    if (dtoHashMap.containsKey(fileName)) {
      result = dtoHashMap.get(fileName);
    }
    return result;
  }


}
